package cn.kennylee.codehub.mq.lip.integration.rocketmq.listener;

import cn.hutool.crypto.digest.DigestUtil;
import lombok.Builder;
import lombok.Setter;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.description.annotation.AnnotationDescription;
import net.bytebuddy.dynamic.DynamicType;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * <p> RocketMQMessageListener实现类构建器 </p>
 * <p>Created on 2025/4/17.</p>
 *
 * @author kennylee
 * @since 0.0.1
 */
@Setter
@Accessors(chain = true, fluent = true)
@Builder
@Slf4j
public class RocketMqMessageListenerImplBuilder {

    /**
     * 缓存的RocketMQMessageListener实现类
     */
    private static final ConcurrentMap<String, Class<?>> LISTENER_CLASS_CACHE = new ConcurrentHashMap<>();

    /**
     * 消息主题：一般建议使用常量，并且单个微服务主题数量为1-3个。
     */
    private final String topic;

    /**
     * 消息过滤选择器，解析语法受 {@link SelectorType} 影响
     */
    private final String selectorExpression;

    /**
     * 消费组；实测RocketMQ下一个消费者一个消费组，同Topic下不能重复，否则消费会有异常
     */
    private final String consumerGroup;

    /**
     * 消息选择表达式类型
     */
    @Builder.Default
    private SelectorType selectorType = SelectorType.TAG;

    /**
     * 设置同1时间最大消费线程数
     *
     * @see RocketMQMessageListener#consumeThreadMax()
     */
    @Builder.Default
    private int consumeThreadMax = 64;

    /**
     * 每个线程消费消息数
     * <p>注：如果consumeThreadMax和consumeThreadNumber都是1就可以单队列，不担心并发问题了</p>
     *
     * @see RocketMQMessageListener#consumeThreadNumber()
     */
    @Builder.Default
    private int consumeThreadNumber = 20;

    /**
     * <p>消息模型</p>
     * <ol>
     *     <li>CLUSTERING 集群模式 :一条消息只会被消费者集群中的一个实例消费 </li>
     *     <li>BROADCASTING 广播模式: 一条消息会被所有消费者实例消费 </li>
     * </ol>
     */
    @Builder.Default
    private MessageModel messageModel = MessageModel.CLUSTERING;

    /**
     * <p>消费模式</p>
     * <ol>
     *     <li>CONCURRENTLY 并发消费: 多线程并发消费消息 </li>
     *     <li>ORDERLY 顺序消费: 队列串行消费消息；注，不能保证不发发生并发</li>
     * </ol>
     */
    @Builder.Default
    private ConsumeMode consumeMode = ConsumeMode.CONCURRENTLY;

    @Builder.Default
    private String tlsEnable = "false";

    /**
     * RocketMqListenerConfigBuilder静态构建器
     *
     * @param topic              消息主题
     * @param selectorExpression 消息标签
     * @param consumerGroup      消费组；实测RocketMQ下一个消费者一个消费组，同Topic下不能重复，否则消费会有异常
     * @return RocketMqListenerConfigBuilder
     */
    public static RocketMqMessageListenerImplBuilder of(String topic, String consumerGroup, String selectorExpression) {
        return RocketMqMessageListenerImplBuilder.builder()
            .topic(topic)
            .selectorExpression(selectorExpression)
            .consumerGroup(consumerGroup)
            .build();
    }

    /**
     * 动态创建一个实现了 RocketMQMessageListener 接口的类
     *
     * @return RocketMQMessageListener 实现类
     */
    public RocketMQMessageListener build() {
        // 业务key
        String rawKey = consumerGroup + "#" + topic + "#" + selectorExpression;
        // 避免特殊字符导致的缓存key不合法，生成一个MD5值作为缓存key
        final String cacheKey = DigestUtil.md5Hex(rawKey);

        final Class<?> listenerClass = LISTENER_CLASS_CACHE.computeIfAbsent(cacheKey, k -> {
            try (DynamicType.Unloaded<Object> unloaded = new ByteBuddy()
                .subclass(Object.class)
                .implement(RocketMQMessageListener.class)
                .annotateType(AnnotationDescription.Builder.ofType(RocketMQMessageListener.class)
                    .define("consumerGroup", consumerGroup)
                    .define("topic", topic)
                    .define("selectorExpression", selectorExpression)
                    .define("selectorType", selectorType)
                    .define("consumeThreadMax", consumeThreadMax)
                    .define("consumeThreadNumber", consumeThreadNumber)
                    .define("messageModel", messageModel)
                    .define("consumeMode", consumeMode)
                    .define("tlsEnable", tlsEnable)
                    .build())
                .make()) {
                return unloaded
                    .load(getClass().getClassLoader())
                    .getLoaded();
            }
        });

        return (RocketMQMessageListener) listenerClass.getDeclaredAnnotations()[0];
    }
}
